package org.jetbrains.exposed.v1.core

import org.jetbrains.exposed.v1.core.dao.id.EntityID
import org.jetbrains.exposed.v1.core.dao.id.EntityIDFunctionProvider
import org.jetbrains.exposed.v1.core.dao.id.IdTable
import org.jetbrains.exposed.v1.core.statements.api.ExposedBlob
import org.jetbrains.exposed.v1.core.statements.api.PreparedStatementApi
import org.jetbrains.exposed.v1.core.statements.api.RowApi
import org.jetbrains.exposed.v1.core.transactions.currentTransaction
import org.jetbrains.exposed.v1.core.vendors.H2Dialect
import org.jetbrains.exposed.v1.core.vendors.MariaDBDialect
import org.jetbrains.exposed.v1.core.vendors.MysqlDialect
import org.jetbrains.exposed.v1.core.vendors.OracleDialect
import org.jetbrains.exposed.v1.core.vendors.PostgreSQLDialect
import org.jetbrains.exposed.v1.core.vendors.PostgreSQLNGDialect
import org.jetbrains.exposed.v1.core.vendors.SQLServerDialect
import org.jetbrains.exposed.v1.core.vendors.currentDialect
import org.jetbrains.exposed.v1.core.vendors.h2Mode
import java.io.InputStream
import java.math.BigDecimal
import java.math.BigInteger
import java.math.MathContext
import java.math.RoundingMode
import java.nio.ByteBuffer
import java.sql.Blob
import java.sql.Clob
import java.util.*
import kotlin.reflect.KClass
import kotlin.reflect.full.isSubclassOf

/**
 * Interface common to all column types.
 */
@Suppress("TooManyFunctions")
interface IColumnType<T> {
    /** Returns `true` if the column type is nullable, `false` otherwise. */
    var nullable: Boolean

    /** Returns the SQL type of this column. */
    fun sqlType(): String

    /**
     * Converts the specified [value] (from the database) to an object of the appropriated type, for this column type.
     * Default implementation returns the same instance.
     */
    fun valueFromDB(value: Any): T?

    /** Returns an object compatible with the database, from the specified [value], for this column type. */
    fun valueToDB(value: T?): Any? = value?.let(::notNullValueToDB)

    /** Returns an object compatible with the database, from the specified **non-null** [value], for this column type. */
    fun notNullValueToDB(value: T & Any): Any = value

    /**
     * Returns the SQL representation of the specified [value], for this column type.
     * If the value is `null` and the column is not nullable, an exception will be thrown.
     * Used when generating an SQL statement and when logging that statement.
     */
    fun valueToString(value: T?): String = when (value) {
        null -> {
            check(nullable) { "NULL in non-nullable column" }
            "NULL"
        }

        else -> nonNullValueToString(value)
    }

    /** Returns the SQL representation of the specified **non-null** [value], for this column type. */
    fun nonNullValueToString(value: T & Any): String = notNullValueToDB(value).toString()

    /**
     * Returns the String representation of the specified [value] when [value] is set as the default for
     * the column.
     * If the value is `null` and the column is not nullable, an exception will be thrown.
     * Used for metadata default value comparison.
     */
    fun valueAsDefaultString(value: T?): String = when (value) {
        null -> {
            check(nullable) { "NULL in non-nullable column" }
            "NULL"
        }

        else -> nonNullValueAsDefaultString(value)
    }

    /**
     * Returns the String representation of the specified **non-null** [value] when [value] is set as the default for
     * the column.
     */
    fun nonNullValueAsDefaultString(value: T & Any): String = nonNullValueToString(value)

    /** Returns the object at the specified [index] in the [rs]. */
    fun readObject(rs: RowApi, index: Int): Any? = rs.getObject(index, null, this)

    /** Sets the [value] at the specified [index] into the [stmt]. */
    fun setParameter(stmt: PreparedStatementApi, index: Int, value: Any?) {
        if (value == null || value is Op.NULL) {
            stmt.setNull(index, this)
        } else {
            stmt.set(index, value, this)
        }
    }

    /**
     * Function checks that provided value is suites the column type and throws [IllegalArgumentException] otherwise.
     * [value] can be of any type (including [Expression])
     * */
    @Throws(IllegalArgumentException::class)
    fun validateValueBeforeUpdate(value: T?) {
    }

    /**
     * Defines the appearance of parameter markers in prepared SQL statements.
     *
     * The default parameter marker is `?`, but it can be overridden in specific cases.
     *
     * For example, H2 uses `? FORMAT JSON` for JSON columns,
     * in Postgres a parameter marker can be explicitly cast to a specific type, etc.
     */
    fun parameterMarker(value: T?) = "?"
}

/**
 * Standard column type.
 */
abstract class ColumnType<T>(override var nullable: Boolean = false) : IColumnType<T> {
    override fun toString(): String = sqlType()
    override fun equals(other: Any?): Boolean {
        if (this === other) return true
        if (javaClass != other?.javaClass) return false

        other as ColumnType<*>

        return nullable == other.nullable
    }

    override fun hashCode(): Int = 31 * javaClass.hashCode() + nullable.hashCode()
}

/**
 * Auto-increment column type.
 */
class AutoIncColumnType<T>(
    /** Returns the base column type of this auto-increment column. */
    val delegate: ColumnType<T>,
    private val _autoincSeq: String?,
    private val fallbackSeqName: String
) : IColumnType<T> by delegate {

    private var _sequence: Sequence? = null

    /** The sequence used to generate new values for this auto-increment column. */
    val sequence: Sequence?
        get() = _sequence ?: autoincSeq?.let {
            Sequence(
                it,
                startWith = 1,
                minValue = 1,
                maxValue = currentDialect.sequenceMaxValue
            )
        }

    constructor(delegate: ColumnType<T>, sequence: Sequence) : this(delegate, sequence.name, sequence.name) {
        _sequence = sequence
    }

    /** The name of the sequence used to generate new values for this auto-increment column. */
    val autoincSeq: String?
        get() = _autoincSeq.takeIf { currentDialect.supportsCreateSequence }
            ?: fallbackSeqName.takeIf { currentDialect.needsSequenceToAutoInc }

    /** The SQL expression that advances the sequence of this auto-increment column. */
    val nextValExpression: NextVal<*>?
        get() = autoincSeq?.let {
            if (delegate is IntegerColumnType) sequence?.nextIntVal() else sequence?.nextLongVal()
        }

    private fun resolveAutoIncType(columnType: IColumnType<*>): String = when (columnType) {
        is EntityIDColumnType<*> -> resolveAutoIncType(columnType.idColumn.columnType)
        is IntegerColumnType if autoincSeq != null -> currentDialect.dataTypeProvider.integerType()
        is IntegerColumnType -> currentDialect.dataTypeProvider.integerAutoincType()
        is LongColumnType if autoincSeq != null -> currentDialect.dataTypeProvider.longType()
        is LongColumnType -> currentDialect.dataTypeProvider.longAutoincType()
        else -> guessAutoIncTypeBy(columnType.sqlType())
    } ?: error("Unsupported type $delegate for auto-increment")

    private fun guessAutoIncTypeBy(sqlType: String): String? = when (sqlType) {
        currentDialect.dataTypeProvider.longType() -> currentDialect.dataTypeProvider.longAutoincType()
        currentDialect.dataTypeProvider.integerType() -> currentDialect.dataTypeProvider.integerAutoincType()
        currentDialect.dataTypeProvider.ulongType() -> currentDialect.dataTypeProvider.ulongAutoincType()
        currentDialect.dataTypeProvider.uintegerType() -> currentDialect.dataTypeProvider.uintegerAutoincType()
        else -> null
    }

    override fun sqlType(): String = resolveAutoIncType(delegate)

    override fun equals(other: Any?): Boolean {
        return when {
            other == null -> false
            this === other -> true
            this::class != other::class -> false
            other !is AutoIncColumnType<*> -> false
            delegate != other.delegate -> false
            _autoincSeq != other._autoincSeq -> false
            fallbackSeqName != other.fallbackSeqName -> false
            sequence != other.sequence -> false
            else -> true
        }
    }

    override fun hashCode(): Int {
        var result = delegate.hashCode()
        result = 31 * result + (_autoincSeq?.hashCode() ?: 0)
        result = 31 * result + fallbackSeqName.hashCode()
        result = 31 * result + (_sequence?.hashCode() ?: 0)
        return result
    }
}

/** Returns `true` if this is an auto-increment column, `false` otherwise. */
val IColumnType<*>.isAutoInc: Boolean
    get() = this is AutoIncColumnType || (this is EntityIDColumnType<*> && idColumn.columnType.isAutoInc)

/** Returns this column's type cast as [AutoIncColumnType] or `null` if the cast fails. */
val Column<*>.autoIncColumnType: AutoIncColumnType<*>?
    get() = (columnType as? AutoIncColumnType)
        ?: (columnType as? EntityIDColumnType<*>)?.idColumn?.columnType as? AutoIncColumnType

internal fun IColumnType<*>.rawSqlType(): IColumnType<*> = when (this) {
    is AutoIncColumnType -> delegate
    is EntityIDColumnType<*> if idColumn.columnType is AutoIncColumnType -> idColumn.columnType.delegate
    else -> this
}

/** Returns whether this column is registered to an [IdTable] and is that table's `id` column. */
internal fun Column<*>.isEntityIdentifier(): Boolean {
    if (columnType !is EntityIDColumnType<*>) return false

    val tableToCheck = ((table as? Alias<*>)?.delegate ?: table) as? IdTable<*>
    return tableToCheck?.id == columnType.idColumn
}

/**
 * Identity column type for storing unique [EntityID] values.
 */
class EntityIDColumnType<T : Any>(
    /** The underlying wrapped column storing the identity values. */
    val idColumn: Column<T>
) : ColumnType<EntityID<T>>() {

    init {
        require(idColumn.table is IdTable<*>) { "EntityId supported only for IdTables" }
    }

    override fun sqlType(): String = idColumn.columnType.sqlType()

    override fun notNullValueToDB(value: EntityID<T>): Any = idColumn.columnType.notNullValueToDB(value.value)

    override fun nonNullValueToString(value: EntityID<T>): String = idColumn.columnType.nonNullValueToString(value.value)

    @Suppress("UNCHECKED_CAST")
    override fun valueFromDB(value: Any): EntityID<T> = EntityIDFunctionProvider.createEntityID(
        when (value) {
            is EntityID<*> -> value.value as T
            else -> idColumn.columnType.valueFromDB(value) as T
        },
        idColumn.table as IdTable<T>
    )

    override fun readObject(rs: RowApi, index: Int): Any? = idColumn.columnType.readObject(rs, index)

    override fun equals(other: Any?): Boolean {
        if (other !is EntityIDColumnType<*>) return false

        return idColumn == other.idColumn
    }

    override fun hashCode(): Int = 31 * super.hashCode() + idColumn.hashCode()
}

/**
 * An interface defining the transformation between a source column type and a target type.
 *
 * @param Wrapped The type of the column values after transformation
 * @param Unwrapped The type of the column values without transformation
 */
interface ColumnTransformer<Unwrapped, Wrapped> {
    /**
     * Returns the underlying column value without a transformation applied ([Wrapped] -> [Unwrapped]).
     */
    fun unwrap(value: Wrapped): Unwrapped

    /**
     * Applies transformation to the underlying column value ([Unwrapped] -> [Wrapped])
     */
    fun wrap(value: Unwrapped): Wrapped
}

fun <Unwrapped, Wrapped> columnTransformer(
    unwrap: (value: Wrapped) -> Unwrapped,
    wrap: (value: Unwrapped) -> Wrapped
): ColumnTransformer<Unwrapped, Wrapped> {
    return object : ColumnTransformer<Unwrapped, Wrapped> {
        override fun unwrap(value: Wrapped): Unwrapped = unwrap(value)
        override fun wrap(value: Unwrapped): Wrapped = wrap(value)
    }
}

/**
 * A class that handles the transformation between a source column type and a target type.
 *
 * [ColumnWithTransform] extends [ColumnType] and can be used to define columns.
 *
 * It does not apply transformations to `null` values. `null` values are passed directly
 * to the delegated column type. If you need to transform `null` values, use [NullableColumnWithTransform] instead.
 *
 * @param Wrapped The type to which the column value of type [Unwrapped] is transformed
 * @param Unwrapped The type of the column
 * @param delegate The original column's [IColumnType]
 * @param transformer Instance of [ColumnTransformer] that makes actual data transformation
 */
open class ColumnWithTransform<Unwrapped, Wrapped>(
    val delegate: IColumnType<Unwrapped & Any>,
    val transformer: ColumnTransformer<Unwrapped, Wrapped>
) : ColumnType<Wrapped & Any>() {

    /**
     * Recursively unwraps the given value by applying the delegate's transformation.
     *
     * This method will recursively call unwrap on the inner delegate if the delegate
     * is also an instance of [ColumnWithTransform]. This is useful for handling nested
     * transformations.
     *
     * @param value The value to unwrap. Could be null.
     * @return The unwrapped value. Returns the value transformed by the transformer if it's not null.
     */
    open fun unwrapRecursive(value: Wrapped?): Any? {
        return if (delegate is ColumnWithTransform<*, *>) {
            (delegate as ColumnWithTransform<Any, Unwrapped>).unwrapRecursive(transformer.unwrap(value as Wrapped))
        } else {
            value?.let { transformer.unwrap(value) }
        }
    }

    /**
     * Gets the original column type that this column with transformation wraps around.
     *
     * This property will recursively unwrap the delegate column type if the delegate
     * is also an instance of [ColumnWithTransform]. This ensures that you get the
     * original column type, regardless of the number of nested transformations.
     *
     * @return The original column's [IColumnType].
     */
    val originalColumnType: IColumnType<Any>
        get() = when {
            delegate is ColumnWithTransform<*, *> -> delegate.originalColumnType
            else -> delegate as IColumnType<Any>
        }

    override fun sqlType(): String = delegate.sqlType()

    override var nullable: Boolean
        get() = delegate.nullable
        set(value) {
            delegate.nullable = value
        }

    override fun valueFromDB(value: Any): Wrapped? {
        return delegate.valueFromDB(value)?.let { transformer.wrap(it) }
    }

    override fun valueToDB(value: Wrapped?): Any? {
        return delegate.valueToDB(value?.let { transformer.unwrap(it) })
    }

    override fun notNullValueToDB(value: Wrapped & Any): Any {
        return delegate.notNullValueToDB(transformer.unwrap(value)!!)
    }

    override fun readObject(rs: RowApi, index: Int): Any? = delegate.readObject(rs, index)

    override fun setParameter(stmt: PreparedStatementApi, index: Int, value: Any?) {
        return delegate.setParameter(stmt, index, value)
    }

    override fun parameterMarker(value: Wrapped?): String {
        return delegate.parameterMarker(value?.let { transformer.unwrap(it) })
    }
}

/**
 * A class that handles the transformation between a source column type and a target type,
 * but also supports transformations involving `null` values.
 *
 * [NullableColumnWithTransform] extends [ColumnType] and can be used to define columns
 * that can transform `null` to `non-null` values and vice versa.
 *
 * @param Wrapped The type to which the column value of type [Unwrapped] is transformed
 * @param Unwrapped The type of the column
 * @param delegate The original column's [IColumnType]
 * @param transformer Instance of [ColumnTransformer] that makes actual data transformation
 */
open class NullableColumnWithTransform<Unwrapped, Wrapped>(
    delegate: IColumnType<Unwrapped & Any>,
    transformer: ColumnTransformer<Unwrapped, Wrapped>
) : ColumnWithTransform<Unwrapped, Wrapped>(delegate, transformer) {
    /**
     * Recursively unwraps the given value by applying the delegate's transformation.
     *
     * This method will recursively call unwrap on the inner delegate if the delegate
     * is also an instance of [ColumnWithTransform]. This is useful for handling nested
     * transformations. Unlike [ColumnWithTransform.unwrapRecursive], this method allows
     * transformation involving `null` values.
     *
     * @param value The value to unwrap. Could be `null`.
     * @return The unwrapped value. Returns the value transformed by the transformer, which
     * could be `null` if the transformer design allows it.
     */
    override fun unwrapRecursive(value: Wrapped?): Any? {
        return if (delegate is ColumnWithTransform<*, *>) {
            (delegate as ColumnWithTransform<Any, Unwrapped>).unwrapRecursive(transformer.unwrap(value as Wrapped))
        } else {
            transformer.unwrap(value as Wrapped)
        }
    }

    override fun valueFromDB(value: Any): Wrapped? {
        return transformer.wrap(delegate.valueFromDB(value) as Unwrapped)
    }

    override fun valueToDB(value: Wrapped?): Any? {
        // All the values go through the transformer since it can map null values to non-null values
        return delegate.valueToDB(transformer.unwrap(value as Wrapped))
    }

    override fun valueToString(value: Wrapped?): String {
        return delegate.valueToString(transformer.unwrap(value as Wrapped))
    }
}

// Numeric columns

/**
 * Numeric column for storing 1-byte integers.
 */
class ByteColumnType : ColumnType<Byte>() {
    override fun sqlType(): String = currentDialect.dataTypeProvider.byteType()

    override fun valueFromDB(value: Any): Byte = when (value) {
        is Byte -> value
        is Number -> value.toByte()
        is String -> value.toByte()
        else -> error("Unexpected value of type Byte: $value of ${value::class.qualifiedName}")
    }

    override fun valueToDB(value: Byte?): Any? {
        return if (currentDialect is SQLServerDialect) {
            // Workaround for SQL Server JDBC driver mysterious error for in-range values if there's a CHECK constraint
            value?.toShort()
        } else {
            super.valueToDB(value)
        }
    }
}

/**
 * Numeric column for storing unsigned 1-byte integers.
 *
 * **Note:** If the database being used is not MySQL, MariaDB, or SQL Server, this column will represent the
 * database's 2-byte integer type with a check constraint that ensures storage of only values
 * between 0 and [UByte.MAX_VALUE] inclusive.
 */
class UByteColumnType : ColumnType<UByte>() {
    override fun sqlType(): String = currentDialect.dataTypeProvider.ubyteType()

    override fun valueFromDB(value: Any): UByte {
        return when (value) {
            is UByte -> value
            is Byte -> value.toUByte()
            is Number -> value.toShort().toUByte()
            is String -> value.toUByte()
            else -> error("Unexpected value of type Byte: $value of ${value::class.qualifiedName}")
        }
    }

    override fun setParameter(stmt: PreparedStatementApi, index: Int, value: Any?) {
        val v = when (value) {
            is UByte -> value.toShort()
            else -> value
        }
        super.setParameter(stmt, index, v)
    }

    override fun notNullValueToDB(value: UByte): Any = value.toShort()
}

/**
 * Numeric column for storing 2-byte integers.
 */
class ShortColumnType : ColumnType<Short>() {
    override fun sqlType(): String = currentDialect.dataTypeProvider.shortType()
    override fun valueFromDB(value: Any): Short = when (value) {
        is Short -> value
        is Number -> value.toShort()
        is String -> value.toShort()
        else -> error("Unexpected value of type Short: $value of ${value::class.qualifiedName}")
    }
}

/**
 * Numeric column for storing unsigned 2-byte integers.
 *
 * **Note:** If the database being used is not MySQL or MariaDB, this column will represent the database's 4-byte
 * integer type with a check constraint that ensures storage of only values between 0 and [UShort.MAX_VALUE] inclusive.
 */
class UShortColumnType : ColumnType<UShort>() {
    override fun sqlType(): String = currentDialect.dataTypeProvider.ushortType()
    override fun valueFromDB(value: Any): UShort {
        return when (value) {
            is UShort -> value
            is Short -> value.toUShort()
            is Number -> value.toInt().toUShort()
            is String -> value.toUShort()
            else -> error("Unexpected value of type Short: $value of ${value::class.qualifiedName}")
        }
    }

    override fun setParameter(stmt: PreparedStatementApi, index: Int, value: Any?) {
        val v = when (value) {
            is UShort -> value.toInt()
            else -> value
        }
        super.setParameter(stmt, index, v)
    }

    override fun notNullValueToDB(value: UShort): Any = value.toInt()
}

/**
 * Numeric column for storing 4-byte integers.
 */
class IntegerColumnType : ColumnType<Int>() {
    override fun sqlType(): String = currentDialect.dataTypeProvider.integerType()
    override fun valueFromDB(value: Any): Int = when (value) {
        is Int -> value
        is Number -> value.toInt()
        is String -> value.toInt()
        else -> error("Unexpected value of type Int: $value of ${value::class.qualifiedName}")
    }
}

/**
 * Numeric column for storing unsigned 4-byte integers.
 *
 * **Note:** If the database being used is not MySQL or MariaDB, this column will use the database's
 * 8-byte integer type with a check constraint that ensures storage of only values
 * between 0 and [UInt.MAX_VALUE] inclusive.
 */
class UIntegerColumnType : ColumnType<UInt>() {
    override fun sqlType(): String = currentDialect.dataTypeProvider.uintegerType()
    override fun valueFromDB(value: Any): UInt {
        return when (value) {
            is UInt -> value
            is Int -> value.toUInt()
            is Number -> value.toLong().toUInt()
            is String -> value.toUInt()
            else -> error("Unexpected value of type Int: $value of ${value::class.qualifiedName}")
        }
    }

    override fun setParameter(stmt: PreparedStatementApi, index: Int, value: Any?) {
        val v = when (value) {
            is UInt -> value.toLong()
            else -> value
        }
        super.setParameter(stmt, index, v)
    }

    override fun notNullValueToDB(value: UInt): Any = value.toLong()
}

/**
 * Numeric column for storing 8-byte integers.
 */
class LongColumnType : ColumnType<Long>() {
    override fun sqlType(): String = currentDialect.dataTypeProvider.longType()
    override fun valueFromDB(value: Any): Long = when (value) {
        is Long -> value
        is Number -> value.toLong()
        is String -> value.toLong()
        else -> error("Unexpected value of type Long: $value of ${value::class.qualifiedName}")
    }
}

/**
 * Numeric column for storing unsigned 8-byte integers.
 */
class ULongColumnType : ColumnType<ULong>() {
    override fun sqlType(): String = currentDialect.dataTypeProvider.ulongType()
    override fun valueFromDB(value: Any): ULong {
        return when (value) {
            is ULong -> value
            is Long -> value.takeIf { it >= 0 }?.toULong()
            is Double -> value.takeIf { it >= 0 }?.toULong() // For SQLite
            is Number -> {
                valueFromDB(value.toString())
            }

            is String -> {
                value.toBigInteger().takeIf {
                    it >= "0".toBigInteger() && it <= ULong.MAX_VALUE.toString().toBigInteger()
                }?.toString()?.toULong()
            }

            else -> error("Unexpected value of type Long: $value of ${value::class.qualifiedName}")
        } ?: error("Negative value but type is ULong: $value")
    }

    override fun notNullValueToDB(value: ULong): Any {
        return when (currentDialect) {
            // PostgreSQLNG does not throw `out of range` error, so it's handled here to prevent storing invalid values
            is PostgreSQLNGDialect -> {
                value.takeIf { it >= 0uL && it <= Long.MAX_VALUE.toULong() }?.toLong()
                    ?: error("Value out of range: $value")
            }
            is PostgreSQLDialect -> BigInteger(value.toString())
            // Long is also an accepted mapping, but this would require handling as above for Oor errors
            is H2Dialect -> BigDecimal(value.toString())
            else -> value.toString()
        }
    }
}

/**
 * Numeric column for storing 4-byte (single precision) floating-point numbers.
 */
class FloatColumnType : ColumnType<Float>() {
    override fun sqlType(): String = currentDialect.dataTypeProvider.floatType()
    override fun valueFromDB(value: Any): Float = when (value) {
        is Float -> value
        is Number -> value.toFloat()
        is String -> value.toFloat()
        else -> error("Unexpected value of type Float: $value of ${value::class.qualifiedName}")
    }

    override fun nonNullValueAsDefaultString(value: Float): String {
        return value.toString().let {
            when {
                // MySQL returns floating-point numbers from metadata without a decimal part as integer strings, whereas other databases
                // append a trailing zero.
                // For example, the value 30f would be `"30"` in MySQL but `"30.0"` in other databases.
                currentDialect is MysqlDialect && it.endsWith(".0") -> it.replace(".0", "")
                else -> it
            }
        }
    }
}

/**
 * Numeric column for storing 8-byte (double precision) floating-point numbers.
 */
class DoubleColumnType : ColumnType<Double>() {
    override fun sqlType(): String = currentDialect.dataTypeProvider.doubleType()
    override fun valueFromDB(value: Any): Double = when (value) {
        is Double -> value
        // Cast as string to prevent precision loss
        is Float -> value.toString().toDouble()
        is Number -> value.toDouble()
        is String -> value.toDouble()
        else -> error("Unexpected value of type Double: $value of ${value::class.qualifiedName}")
    }

    override fun nonNullValueAsDefaultString(value: Double): String {
        return value.toString().let {
            when {
                // MySQL returns floating-point numbers from metadata without a decimal part as integer strings, whereas other databases
                // append a trailing zero.
                // For example, the value 30f would be `"30"` in MySQL but `"30.0"` in other databases.
                currentDialect is MysqlDialect && it.endsWith(".0") -> it.replace(".0", "")
                else -> it
            }
        }
    }
}

/**
 * Numeric column for storing numbers with the specified [precision] and [scale].
 */
class DecimalColumnType(
    /** Total count of significant digits in the whole number. */
    val precision: Int,
    /** Count of decimal digits in the fractional part. */
    val scale: Int
) : ColumnType<BigDecimal>() {
    override fun sqlType(): String = "DECIMAL($precision, $scale)"

    override fun valueFromDB(value: Any): BigDecimal = when (value) {
        is BigDecimal -> value
        is Double -> {
            if (value.isNaN()) {
                error("Unexpected value of type Double: NaN of ${value::class.qualifiedName}")
            } else {
                value.toBigDecimal()
            }
        }

        is Float -> {
            if (value.isNaN()) {
                error("Unexpected value of type Float: NaN of ${value::class.qualifiedName}")
            } else {
                value.toBigDecimal()
            }
        }

        is Long -> value.toBigDecimal()
        is Int -> value.toBigDecimal()
        is Short -> value.toLong().toBigDecimal()
        else -> error("Unexpected value of type Decimal: $value of ${value::class.qualifiedName}")
    }.setScale(scale, RoundingMode.HALF_EVEN)

    override fun equals(other: Any?): Boolean {
        if (this === other) return true
        if (javaClass != other?.javaClass) return false
        if (!super.equals(other)) return false

        other as DecimalColumnType

        if (precision != other.precision) return false
        if (scale != other.scale) return false

        return true
    }

    override fun hashCode(): Int {
        var result = super.hashCode()
        result = 31 * result + precision
        result = 31 * result + scale
        return result
    }

    companion object {
        internal val INSTANCE = DecimalColumnType(MathContext.DECIMAL64.precision, 20)
    }
}

// Character columns

/**
 * Character column for storing single characters.
 */
class CharacterColumnType : ColumnType<Char>() {
    override fun sqlType(): String = "CHAR"
    override fun valueFromDB(value: Any): Char = when (value) {
        is Char -> value
        is Number -> value.toInt().toChar()
        is String -> value.single()
        else -> error("Unexpected value of type Char: $value of ${value::class.qualifiedName}")
    }

    override fun notNullValueToDB(value: Char): Any = value.toString()

    override fun nonNullValueToString(value: Char): String = "'$value'"
}

/**
 * Base character column for storing strings using the specified text [collate] type.
 */
abstract class StringColumnType(
    /** Returns the collate type used in by this column. */
    val collate: String? = null
) : ColumnType<String>() {
    /** Returns the specified [value] with special characters escaped. */
    protected fun escape(value: String): String = value.map { charactersToEscape[it] ?: it }.joinToString("")

    /** Returns the specified [value] with special characters escaped and wrapped in quotations, if necessary. */
    protected fun escapeAndQuote(value: String): String = when (currentDialect) {
        is PostgreSQLDialect -> "\"${escape(value)}\""
        else -> escape(value)
    }

    override fun valueFromDB(value: Any): String = when (value) {
        is Clob -> value.characterStream.readText()
        is ByteArray -> String(value)
        else -> value.toString()
    }

    override fun nonNullValueToString(value: String): String = buildString {
        append('\'')
        append(escape(value))
        append('\'')
    }

    override fun equals(other: Any?): Boolean {
        if (this === other) return true
        if (javaClass != other?.javaClass) return false
        if (!super.equals(other)) return false

        other as StringColumnType

        return collate == other.collate
    }

    override fun hashCode(): Int {
        var result = super.hashCode()
        result = 31 * result + (collate?.hashCode() ?: 0)
        return result
    }

    companion object {
        private val charactersToEscape = mapOf(
            '\'' to "\'\'",
            '\r' to "\\r",
            '\n' to "\\n"
        )
    }
}

/**
 * Character column for storing strings with the exact [colLength] length using the specified [collate] type.
 */
open class CharColumnType(
    /** Returns the maximum length of this column. */
    val colLength: Int = 255,
    collate: String? = null
) : StringColumnType(collate) {
    override fun sqlType(): String = buildString {
        append("CHAR($colLength)")
        if (collate != null) {
            append(" COLLATE ${escapeAndQuote(collate)}")
        }
    }

    override fun validateValueBeforeUpdate(value: String?) {
        if (value is String) {
            val valueLength = value.codePointCount(0, value.length)
            require(valueLength <= colLength) {
                "Value can't be stored to database column because exceeds length ($valueLength > $colLength)"
            }
        }
    }

    override fun equals(other: Any?): Boolean {
        if (this === other) return true
        if (javaClass != other?.javaClass) return false
        if (!super.equals(other)) return false

        other as CharColumnType

        if (colLength != other.colLength) return false

        if (collate != other.collate) return false
        return true
    }

    override fun hashCode(): Int {
        var result = super.hashCode()
        result = 31 * result + colLength
        return result
    }
}

/**
 * Character column for storing strings with the specified maximum [colLength] using the specified [collate] type.
 */
open class VarCharColumnType(
    /** Returns the maximum length of this column. */
    val colLength: Int = 255,
    collate: String? = null
) : StringColumnType(collate) {
    open fun preciseType() = currentDialect.dataTypeProvider.varcharType(colLength)

    override fun sqlType(): String = buildString {
        append(preciseType())
        if (collate != null) {
            append(" COLLATE ${escapeAndQuote(collate)}")
        }
    }

    override fun validateValueBeforeUpdate(value: String?) {
        if (value is String) {
            val valueLength = value.codePointCount(0, value.length)
            require(valueLength <= colLength) {
                "Value can't be stored to database column because exceeds length ($valueLength > $colLength)"
            }
        }
    }

    override fun equals(other: Any?): Boolean {
        if (this === other) return true
        if (javaClass != other?.javaClass) return false
        if (!super.equals(other)) return false

        other as VarCharColumnType

        return colLength == other.colLength
    }

    override fun hashCode(): Int {
        var result = super.hashCode()
        result = 31 * result + colLength
        return result
    }
}

/**
 * Character column for storing strings of arbitrary length using the specified [collate] type.
 */
open class TextColumnType(
    collate: String? = null,
    /** Whether content will be loaded immediately when data is retrieved from the database. */
    val eagerLoading: Boolean = false
) : StringColumnType(collate) {
    /** The exact SQL type representing this character type. */
    open fun preciseType() = currentDialect.dataTypeProvider.textType()

    override fun sqlType(): String = buildString {
        append(preciseType())
        if (collate != null) {
            append(" COLLATE ${escapeAndQuote(collate)}")
        }
    }

    override fun readObject(rs: RowApi, index: Int): Any? {
        val value = super.readObject(rs, index)
        return if (eagerLoading && value != null) {
            valueFromDB(value)
        } else {
            value
        }
    }
}

open class MediumTextColumnType(
    collate: String? = null,
    eagerLoading: Boolean = false
) : TextColumnType(collate, eagerLoading) {
    override fun preciseType(): String = currentDialect.dataTypeProvider.mediumTextType()
}

open class LargeTextColumnType(
    collate: String? = null,
    eagerLoading: Boolean = false
) : TextColumnType(collate, eagerLoading) {
    override fun preciseType(): String = currentDialect.dataTypeProvider.largeTextType()
}

// Binary columns

/**
 * Binary column for storing binary strings of variable and _unlimited_ length.
 */
open class BasicBinaryColumnType : ColumnType<ByteArray>() {
    override fun sqlType(): String = currentDialect.dataTypeProvider.binaryType()

    override fun valueFromDB(value: Any): ByteArray = when (value) {
        is Blob -> value.binaryStream.use { it.readBytes() }
        is InputStream -> value.use { it.readBytes() }
        is ByteArray -> value
        is String -> value.toByteArray()
        is ByteBuffer -> value.array()
        else -> error("Unexpected value $value of type ${value::class.qualifiedName}")
    }

    override fun nonNullValueToString(value: ByteArray): String = value.toString(Charsets.UTF_8)
}

/**
 * Binary column for storing binary strings of a specific [length].
 */
open class BinaryColumnType(
    /** Returns the length of the column- */
    val length: Int
) : BasicBinaryColumnType() {
    override fun sqlType(): String = currentDialect.dataTypeProvider.binaryType(length)

    override fun validateValueBeforeUpdate(value: ByteArray?) {
        if (value is ByteArray) {
            val valueLength = value.size
            require(valueLength <= length) {
                "Value can't be stored to database column because exceeds length ($valueLength > $length)"
            }
        }
    }

    override fun equals(other: Any?): Boolean {
        if (this === other) return true
        if (javaClass != other?.javaClass) return false
        if (!super.equals(other)) return false

        other as BinaryColumnType

        return length == other.length
    }

    override fun hashCode(): Int {
        var result = super.hashCode()
        result = 31 * result + length
        return result
    }
}

/**
 * Binary column for storing BLOBs.
 */
class BlobColumnType(
    /** Returns whether an OID column should be used instead of BYTEA. This value only applies to PostgreSQL databases. */
    val useObjectIdentifier: Boolean = false
) : ColumnType<ExposedBlob>() {
    override fun sqlType(): String = when {
        useObjectIdentifier && currentDialect is PostgreSQLDialect -> "oid"
        useObjectIdentifier -> error("Storing BLOBs using OID columns is only supported by PostgreSQL")
        else -> currentDialect.dataTypeProvider.blobType()
    }

    override fun valueFromDB(value: Any): ExposedBlob = when (value) {
        is ExposedBlob -> value
        is InputStream -> ExposedBlob(value)
        is ByteArray -> ExposedBlob(value)
        is Blob -> ExposedBlob(value.binaryStream)
        is ByteBuffer -> ExposedBlob(value.array())
        else -> error("Unexpected value of type Blob: $value of ${value::class.qualifiedName}")
    }

    override fun nonNullValueToString(value: ExposedBlob): String {
        // For H2 Blobs the original dataTypeProvider must be taken (even if H2 in other DB mode)
        return ((currentDialect as? H2Dialect)?.originalDataTypeProvider ?: currentDialect.dataTypeProvider)
            .hexToDb(value.hexString())
    }

    override fun readObject(rs: RowApi, index: Int) = when {
        currentDialect is PostgreSQLDialect && useObjectIdentifier -> {
            rs.getObject(index, java.sql.Blob::class.java)?.binaryStream?.let(::ExposedBlob)
        }
        else -> rs.getObject(index)
    }

    override fun setParameter(stmt: PreparedStatementApi, index: Int, value: Any?) {
        when (val toSetValue = (value as? ExposedBlob)?.inputStream ?: value) {
            is InputStream -> stmt.setInputStream(index, toSetValue, useObjectIdentifier)
            null, is Op.NULL -> stmt.setNull(index, this)
            else -> super.setParameter(stmt, index, toSetValue)
        }
    }
}

/**
 * Binary column for storing [UUID].
 */
class UUIDColumnType : ColumnType<UUID>() {
    override fun sqlType(): String = currentDialect.dataTypeProvider.uuidType()

    override fun valueFromDB(value: Any): UUID = when (value) {
        is UUID -> value
        is ByteArray -> ByteBuffer.wrap(value).let { b -> UUID(b.long, b.long) }
        is String if value.matches(uuidRegexp) -> UUID.fromString(value)
        is String -> ByteBuffer.wrap(value.toByteArray()).let { b -> UUID(b.long, b.long) }
        is ByteBuffer -> value.let { b -> UUID(b.long, b.long) }
        else -> error("Unexpected value of type UUID: $value of ${value::class.qualifiedName}")
    }

    override fun notNullValueToDB(value: UUID): Any {
        return ((currentDialect as? H2Dialect)?.originalDataTypeProvider ?: currentDialect.dataTypeProvider)
            .uuidToDB(value)
    }

    override fun nonNullValueToString(value: UUID): String = "'$value'"

    @Suppress("MagicNumber")
    override fun readObject(rs: RowApi, index: Int): Any? {
        @OptIn(InternalApi::class)
        val db = currentTransaction().db
        if (currentDialect is MariaDBDialect && !db.version.covers(10)) {
            return rs.getObject(index, java.sql.Array::class.java)
        }
        return super.readObject(rs, index)
    }

    companion object {
        private val uuidRegexp =
            Regex("[0-9A-F]{8}-[0-9A-F]{4}-[0-9A-F]{4}-[0-9A-F]{4}-[0-9A-F]{12}", RegexOption.IGNORE_CASE)
    }
}

// Boolean columns

/**
 * Boolean column for storing boolean values.
 */
class BooleanColumnType : ColumnType<Boolean>() {
    override fun sqlType(): String = currentDialect.dataTypeProvider.booleanType()

    override fun valueFromDB(value: Any): Boolean = when (value) {
        is Number -> value.toLong() != 0L
        is String -> currentDialect.dataTypeProvider.booleanFromStringToBoolean(value)
        else -> value.toString().toBoolean()
    }

    override fun nonNullValueToString(value: Boolean): String =
        currentDialect.dataTypeProvider.booleanToStatementString(value)

    override fun notNullValueToDB(value: Boolean): Any = when {
        (currentDialect is OracleDialect || currentDialect.h2Mode == H2Dialect.H2CompatibilityMode.Oracle) ->
            nonNullValueToString(value)

        else -> value
    }

    companion object {
        internal val INSTANCE = BooleanColumnType()
    }
}

// Enumeration columns

/**
 * Enumeration column for storing enums of type [klass] by their ordinal.
 */
class EnumerationColumnType<T : Enum<T>>(
    /** Returns the enum class used in this column type. */
    val klass: KClass<T>
) : ColumnType<T>() {
    override fun sqlType(): String = currentDialect.dataTypeProvider.integerType()
    private val enumConstants by lazy { klass.java.enumConstants!! }

    @Suppress("UNCHECKED_CAST")
    override fun valueFromDB(value: Any): T = when (value) {
        is Number -> enumConstants[value.toInt()]
        is Enum<*> -> value as T
        else -> error("$value of ${value::class.qualifiedName} is not valid for enum ${klass.simpleName}")
    }

    override fun notNullValueToDB(value: T): Int = value.ordinal

    override fun equals(other: Any?): Boolean {
        if (this === other) return true
        if (javaClass != other?.javaClass) return false
        if (!super.equals(other)) return false

        other as EnumerationColumnType<*>

        return klass == other.klass
    }

    override fun hashCode(): Int {
        var result = super.hashCode()
        result = 31 * result + klass.hashCode()
        return result
    }
}

/**
 * Enumeration column for storing enums of type [klass] by their name.
 */
class EnumerationNameColumnType<T : Enum<T>>(
    /** Returns the enum class used in this column type. */
    val klass: KClass<T>,
    val colLength: Int
) : ColumnType<T>() {
    private val enumConstants by lazy { klass.java.enumConstants!!.associateBy { it.name } }

    override fun sqlType(): String = currentDialect.dataTypeProvider.varcharType(colLength)

    @Suppress("UNCHECKED_CAST")
    override fun valueFromDB(value: Any): T = when (value) {
        is String -> {
            enumConstants[value] ?: error("$value can't be associated with any from enum ${klass.qualifiedName}")
        }

        is Enum<*> -> value as T
        else -> error("$value of ${value::class.qualifiedName} is not valid for enum ${klass.qualifiedName}")
    }

    override fun notNullValueToDB(value: T): Any = value.name

    override fun nonNullValueToString(value: T): String = buildString {
        append('\'')
        append(escape(value.name))
        append('\'')
    }

    override fun validateValueBeforeUpdate(value: T?) {
        if (value != null) {
            val valueLength = value.name.codePointCount(0, value.name.length)
            require(valueLength <= colLength) {
                "Value can't be stored to database column because exceeds length ($valueLength > $colLength)"
            }
        }
    }

    override fun equals(other: Any?): Boolean {
        if (this === other) return true
        if (javaClass != other?.javaClass) return false
        if (!super.equals(other)) return false

        other as EnumerationNameColumnType<*>

        return klass == other.klass
    }

    override fun hashCode(): Int {
        var result = super.hashCode()
        result = 31 * result + klass.hashCode()
        return result
    }

    private fun escape(value: String): String = value.map { charactersToEscape[it] ?: it }.joinToString("")

    companion object {
        private val charactersToEscape = mapOf(
            '\'' to "\'\'",
            '\r' to "\\r",
            '\n' to "\\n"
        )
    }
}

/**
 * Enumeration column for storing enums of type [T] using the custom SQL type [sql].
 */
class CustomEnumerationColumnType<T : Enum<T>>(
    /** Returns the name of this column type instance. */
    val name: String,
    /** Returns the SQL definition used for this column type. */
    val sql: String?,
    /** Returns the function that converts a value received from a database to an enumeration instance [T]. */
    val fromDb: (Any) -> T,
    /** Returns the function that converts an enumeration instance [T] to a value that will be stored to a database. */
    val toDb: (T) -> Any
) : ColumnType<T>() {
    override fun sqlType(): String = sql ?: error("Column $name should exist in database")

    @Suppress("UNCHECKED_CAST")
    override fun valueFromDB(value: Any): T = if (value::class.isSubclassOf(Enum::class)) value as T else fromDb(value)

    override fun notNullValueToDB(value: T): Any = toDb(value)

    override fun nonNullValueToString(value: T): String = buildString {
        append('\'')
        append(escape(notNullValueToDB(value).toString()))
        append('\'')
    }

    private fun escape(value: String): String = value.map { charactersToEscape[it] ?: it }.joinToString("")

    companion object {
        private val charactersToEscape = mapOf(
            '\'' to "\'\'",
            '\r' to "\\r",
            '\n' to "\\n"
        )
    }
}

// Array columns

/**
 * Multi-dimensional array column type for storing a collection of nested elements.
 *
 * @property delegate The base column type associated with this array column's individual elements.
 * @property dimensions The number of dimensions of the multi-dimensional array.
 * @property maximumCardinality The maximum cardinality (number of allowed elements) for each dimension of the array.
 *
 * **Note:** The maximum cardinality is considered for each dimension, but it is ignored by the PostgreSQL database.
 */
class ArrayColumnType<T, R : List<Any?>>(
    val delegate: IColumnType<T & Any>,
    val maximumCardinality: List<Int>? = null,
    val dimensions: Int = 1
) : ColumnType<R>() {

    private val nullElementString = "null"

    /**
     * Constructor with maximum cardinality for a single dimension.
     *
     * @param delegate The base column type associated with this array column's individual elements.
     * @param maximumCardinality The maximum cardinality (number of allowed elements) for the array.
     */
    constructor(delegate: ColumnType<T & Any>, maximumCardinality: Int? = null) : this(delegate, maximumCardinality?.let { listOf(it) })

    /**
     * The SQL type definition of the delegate column type without any potential array dimensions.
     */
    val delegateType: String
        get() = delegate.sqlType().substringBefore('(')

    override fun sqlType(): String = buildString {
        if (maximumCardinality != null) {
            require(maximumCardinality.size == dimensions) {
                "The size of cardinality list must be equal to the amount of array dimensions. " +
                    "Dimensions: $dimensions, cardinality size: ${maximumCardinality.size}"
            }
        }
        append(delegate.sqlType())
        when {
            currentDialect is H2Dialect -> {
                require(dimensions == 1) {
                    "H2 does not support multidimensional arrays. " +
                        "`dimensions` parameter for H2 database must be 1"
                }
                append(" ARRAY", maximumCardinality?.let { "[${it.first()}]" } ?: "")
            }

            else -> append(maximumCardinality?.let { cardinality -> cardinality.joinToString("") { "[$it]" } } ?: "[]".repeat(dimensions))
        }
    }

    override fun notNullValueToDB(value: R): Any {
        return recursiveNotNullValueToDB(value, dimensions)
    }

    private fun recursiveNotNullValueToDB(value: Any, level: Int): Array<Any?> = when {
        level > 1 -> (value as List<Any>).map { recursiveNotNullValueToDB(it, level - 1) }.toTypedArray()
        else -> (value as List<T>).map { it?.let { delegate.notNullValueToDB(it) } }.toTypedArray()
    }

    @Suppress("UNCHECKED_CAST")
    override fun valueFromDB(value: Any): R? {
        return when (value) {
            is Array<*> -> recursiveValueFromDB(value, dimensions) as R?
            is java.sql.Array -> recursiveValueFromDB(value.array, dimensions) as R?
            else -> value as R?
        }
    }

    private fun recursiveValueFromDB(value: Any?, level: Int): List<Any?> = when {
        level > 1 -> (value as Array<Any?>).map { recursiveValueFromDB(it, level - 1) }
        else -> (value as Array<Any?>).map { it?.let { delegate.valueFromDB(it) } }
    }

    override fun readObject(rs: RowApi, index: Int): Any? = rs.getObject(index)

    override fun setParameter(stmt: PreparedStatementApi, index: Int, value: Any?) {
        when (value) {
            is Array<*> if isArrayOfByteArrays(value) ->
                stmt.setArray(index, this, Array(value.size) { value[it] as ByteArray })
            is Array<*> -> stmt.setArray(index, this, value)
            else -> super.setParameter(stmt, index, value)
        }
    }

    override fun nonNullValueToString(value: R): String {
        return arrayLiteralPrefix() + recursiveNonNullValueToString(value, dimensions)
    }

    private fun recursiveNonNullValueToString(value: Any?, level: Int): String = when {
        level > 1 -> (value as List<Any?>).joinToString(",", "[", "]") { recursiveNonNullValueToString(it, level - 1) }
        else -> (value as List<T>).joinToString(",", "[", "]") { it?.let { delegate.nonNullValueToString(it) } ?: nullElementString }
    }

    override fun nonNullValueAsDefaultString(value: R): String {
        return arrayLiteralPrefix() + recursiveNonNullValueAsDefaultString(value, dimensions)
    }

    private fun recursiveNonNullValueAsDefaultString(value: Any?, level: Int): String = when {
        level > 1 -> (value as List<Any?>).joinToString(",", "[", "]") { recursiveNonNullValueAsDefaultString(it, level - 1) }
        else -> (value as List<T>).joinToString(",", "[", "]") { it?.let { delegate.nonNullValueAsDefaultString(it) } ?: nullElementString }
    }

    private fun arrayLiteralPrefix(): String {
        return when {
            currentDialect is H2Dialect -> "ARRAY "
            else -> "ARRAY"
        }
    }

    private fun castH2ParameterMarker(columnType: IColumnType<*>): String? {
        return when (columnType) {
            // Here is the list of types that could be resolved by `resolveColumnType()`.
            // In the common case it must not work for all the possible types. It also does not work with BigDecimal.
            // This cast is needed for array types inside upsert(merge statement), otherwise statement causes "Data conversion error converting" error.
            is ByteColumnType, is UByteColumnType, is BooleanColumnType, is ShortColumnType, is UShortColumnType,
            is IntegerColumnType, is UIntegerColumnType, is LongColumnType, is ULongColumnType, is FloatColumnType,
            is DoubleColumnType, is StringColumnType, is CharacterColumnType, is BasicBinaryColumnType, is UUIDColumnType ->
                "cast(? as ${columnType.sqlType()} array)"
            else -> null
        }
    }

    override fun parameterMarker(value: R?): String {
        if (currentDialect is H2Dialect) {
            val columnType = if (delegate is ColumnWithTransform<*, *>) delegate.originalColumnType else delegate
            return castH2ParameterMarker(columnType) ?: super.parameterMarker(value)
        }

        // For PostgreSQL, add a cast for date arrays to ensure they're properly recognized
        if (currentDialect is PostgreSQLDialect && delegate is IDateColumnType) {
            val pgType = if (delegate.hasTimePart) "timestamp[]" else "date[]"
            return "?::$pgType"
        }

        return super.parameterMarker(value)
    }
}

private fun isArrayOfByteArrays(value: Array<*>) =
    value.all { it is ByteArray }

// Date/Time columns

/**
 * Marker interface for date/datetime related column types.
 **/
interface IDateColumnType {
    val hasTimePart: Boolean
}

// JSON/JSONB columns

/**
 * Marker interface for json/jsonb related column types.
 */
interface JsonColumnMarker {
    val usesBinaryFormat: Boolean
}

/**
 * Returns the [ColumnType] commonly associated with storing values of type [T], or the [defaultType] if a mapping
 * does not exist for type [T].
 *
 * @throws IllegalStateException If no column type mapping is found and a [defaultType] is not provided.
 * @suppress
 */
@InternalApi
fun <T : Any> resolveColumnType(
    klass: KClass<T>,
    defaultType: ColumnType<*>? = null
): ColumnType<T> {
    val type = when (klass) {
        Boolean::class -> BooleanColumnType()
        Byte::class -> ByteColumnType()
        UByte::class -> UByteColumnType()
        Short::class -> ShortColumnType()
        UShort::class -> UShortColumnType()
        Int::class -> IntegerColumnType()
        UInt::class -> UIntegerColumnType()
        Long::class -> LongColumnType()
        ULong::class -> ULongColumnType()
        Float::class -> FloatColumnType()
        Double::class -> DoubleColumnType()
        String::class -> TextColumnType()
        Char::class -> CharacterColumnType()
        ByteArray::class -> BasicBinaryColumnType()
        BigDecimal::class -> DecimalColumnType.INSTANCE
        UUID::class -> UUIDColumnType()
        else -> defaultType
    } as? ColumnType<T>

    return type ?: error(
        "A column type could not be associated with ${klass.qualifiedName}. Provide an explicit column type argument."
    )
}
